/**
 * 
 */
package edu.sjtu.seda.eventqueue;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import edu.sjtu.seda.event.Event;
import edu.sjtu.seda.event.Event.STATUS;
import edu.sjtu.seda.log.LogService;
import edu.sjtu.seda.stage.StageWrapper;
import edu.sjtu.seda.threadpool.StageRunner;
import edu.sjtu.seda.threadpool.StageRunnerImpl;

/**
 * @author john.zhu
 * 
 */
public class EventQueueImpl implements EventQueue {
	private static LogService log = new LogService(EventQueueImpl.class);
	private BlockingQueue<Event> eventQueue;

	private AtomicBoolean shutdown = new AtomicBoolean(false);

	private StageWrapper wrapper;

	private int defaultSize = 10;

	private boolean enableBatch;
	private int batchSize = 10;

	public EventQueueImpl(StageWrapper wrapper, int size) {
		this(wrapper, size, false, 0);
	}

	/**
	 * event queue implementation. Receive events, wrap as stagerunner and send
	 * to threadpool finally send to next stage
	 * 
	 * @param wrapper
	 * @param size
	 */
	public EventQueueImpl(StageWrapper wrapper, int size, boolean enableBatch,
			int batchSize) {
		this.wrapper = wrapper;
		this.defaultSize = size;
		this.enableBatch = enableBatch;
		this.batchSize = batchSize;
		eventQueue = new LinkedBlockingQueue<Event>(defaultSize);
	}

	public void pushEvent(Event event) {
		if (this.eventQueue.offer(event)) {
			log.info("event {0} has been pushed to queue {1}", event.getUID(),
					wrapper.getStage().getName());
		} else {
			event.setStatus(STATUS.REJECTED);
			/*
			 * reject events for the queue is full
			 */
			log.error(MessageFormat.format(
					"Queue {0} is full and event {1} will be rejected.",
					wrapper.getStage().getName(), event.getUID()));
		}
		synchronized (this) {
			notifyAll();
		}
	}

	public Event popEvent() {
		Event event = this.eventQueue.poll();
		log.info(MessageFormat.format(
				"event {0} has been poped from queue {1}", event.getUID(),
				wrapper.getStage().getName()));
		return event;
	}

	public int size() {
		return this.eventQueue.size();
	}

	@Override
	public void clear() {
		this.eventQueue.clear();
	}

	@Override
	public void run() {
		Collection<StageRunner> batchList = new ArrayList<StageRunner>();

		while (true) {

			while (size() != 0) {
				/*
				 * TODO currently queue and threadpool is not synchronized do
				 * some change to support synchronize And pop event in stage
				 * runner
				 */
				Event event = popEvent();
				if (event != null) {
					StageRunner runner = new StageRunnerImpl(event,
							wrapper.getEventHandler(), wrapper);
					if (enableBatch) {
						batchList.add(runner);
						if (batchList.size() >= batchSize) {
							for (StageRunner run : batchList) {
								wrapper.getThreadPool().execute(run);
							}
							batchList = new ArrayList<StageRunner>();
						}
					} else {
						wrapper.getThreadPool().execute(runner);
					}
				} else {
					log.error("unknown error!");
				}
			}
			try {
				synchronized (this) {
					wait();
				}
			} catch (InterruptedException e1) {
				e1.printStackTrace();
			}
		}

	}

	@Override
	public void shutdown() {

	}

	public boolean isShutdown() {
		return size() == 0 && this.shutdown.get();
	}

}
